package com.kingsoft.dc.khaos.module.spark.sink

import com.kingsoft.dc.khaos.{DataSetChannel, KhaosConf, KhaosContext}
import com.kingsoft.dc.khaos.metadata.Dependency
import com.kingsoft.dc.khaos.module.SparkModule
import com.kingsoft.dc.khaos.module.spark.metadata.sink.SinkInfo
import org.apache.spark.sql.DataFrame
import org.json4s._
import org.json4s.jackson.JsonMethods._


class Sink(_mid: String,
           dc: KhaosContext) extends SparkModule {

  override var dependencies: Seq[Dependency] = Seq[Dependency]()
  override var targets: Seq[Dependency] = Seq[Dependency]()
  private var sinkInfo: SinkInfo = _
  private var sinkStrategy: SinkStrategy = _
  private var module_id: String = _

  /**
    *
    * 节点逻辑处理的准备信息
    *
    * @param dittoConf : 组件配置信息
    * @param dc        : 组件上下文环境
    */
  override def prepare(dittoConf: KhaosConf,
                       dc: KhaosContext): this.type = {
//    println("Sink _context is :" + dc.toString)
//    println("Sink _mid is :" + _mid)
    val conf = getModuleConfiguration() match {
      case Some(conf) => conf
      case None => throw new IllegalAccessException("conf is null")
    }
    implicit val formats = DefaultFormats
    sinkInfo = parse(conf, true).extract[SinkInfo]
    module_id = sinkInfo.id
    sinkStrategy =Class.forName(sinkInfo.strategy.clazz,true, Thread.currentThread().getContextClassLoader)
      .newInstance()
      .asInstanceOf[SinkStrategy]

    this
  }

  /**
    * 节点执行逻辑
    *
    * @param dataSetChannel :  数据采集器
    */
  override def execute(dataSetChannel: DataSetChannel): this.type = {
    //KhaosUtils.checkUniqueDependencies(dependencies)
    val (schema, dataSet) = dc.structData[DataFrame](dependencies.head.getEdge())
    //通过接口方式调用具体策略
    sinkStrategy.sink(dc, module_id, sinkInfo.strategy.config, schema, dataSet)
    this
  }

  /**
    * 获取配置信息
    *
    * @return :  返回配置信息
    */
  override def getModuleConfiguration(): Option[String] = {
    dc.conf.getOption(_mid)
  }


  override def addDependencies(dependency: Dependency): this.type = {
    dependencies = dependencies :+ dependency
    this
  }

  def addTarget(dependency: Dependency):this.type ={
    targets = targets :+ dependency
    this
  }
}

